package com.flinksql.test;


import org.apache.flink.api.common.functions.RuntimeContext;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.connectors.elasticsearch.ElasticsearchSinkFunction;
import org.apache.flink.streaming.connectors.elasticsearch.RequestIndexer;
import org.apache.flink.streaming.connectors.elasticsearch7.ElasticsearchSink;

import org.apache.http.HttpHost;
import org.elasticsearch.action.index.IndexRequest;
import org.elasticsearch.client.Requests;

import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;

/**
 * @author: Lin
 * @create: 2021-06-11 17:37
 * @description: 使用ElasticsearchSink.Builder方式，单条写入ES
 **/
public class FlinkSQL_Test5 {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment().setParallelism(1);
        DataStreamSource<String> source = env.socketTextStream("hadoop102", 9999);

        //使用ESBuilder构建输出
        List<HttpHost> hosts = new ArrayList<>();
        hosts.add(new HttpHost("127.0.0.1",9200,"http"));
        ElasticsearchSink.Builder<String> esBuilder = new ElasticsearchSink.Builder<>(hosts, new ElasticsearchSinkFunction<String>() {
            @Override
            public void process(String s, RuntimeContext runtimeContext, RequestIndexer requestIndexer) {

                Map<String, String> jsonMap = new HashMap<>();
                jsonMap.put("data", s);

                IndexRequest indexRequest = Requests.indexRequest();
                indexRequest.index("sensor");
                indexRequest.id("9001");
                indexRequest.source(jsonMap);

                requestIndexer.add(indexRequest);
            }
        });

        esBuilder.setBulkFlushMaxActions(1);//默认批处理方式，这里为了演示使用1
        source.addSink(esBuilder.build());

        env.execute();

    }
}
